package com.zallds.job;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.RestartStrategyOptions;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExactlyOnceOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;

import java.util.Properties;
import java.util.concurrent.TimeUnit;

/**
 * 消费kafka数据到mysql
 * kafka手动提交offset
 */
public class kafka2mysql {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());

        // 一般建议5分钟一次
        env.enableCheckpointing(60000);
        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);

        //设置两个Checkpoint 之间最少等待时间,如设置Checkpoint之间最少是要等 500ms(为了避免每隔1000ms做一次Checkpoint的时候,前一次太慢和后一次重叠到一起去了)
        //如:高速公路上,每隔1s关口放行一辆车,但是规定了两车之前的最小车距为500m
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(30000);

        // Checkpoint 超时时间
        env.getCheckpointConfig().setCheckpointTimeout(60000);

        // 外部检查点 --将数据状态持久化到外部系统中 当作业被取消时，保留外部的checkpoint
        // enable externalized checkpoints which are retained after job cancellation
        env.getCheckpointConfig().enableExternalizedCheckpoints(
                CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);

        //默认值为0，表示不容忍任何检查点失败,允许失败10次
        env.getCheckpointConfig().setTolerableCheckpointFailureNumber(10);

        // enables the experimental unaligned checkpoints
        env.getCheckpointConfig().enableUnalignedCheckpoints();

        // 设置checkPoint位置
        env.getCheckpointConfig().setCheckpointStorage("file:///D:/Destop/movie");

        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(
                3, // 最多重启3次数
                Time.of(5, TimeUnit.SECONDS) // 重启时间间隔
        ));


        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "192.168.62.36:6667");
        properties.setProperty("group.id", "test2");
        properties.setProperty("auto.offset.reset", "earliest");

        //把这个参数设置为false，这个参数的意思会周期性的保存偏移量。
        properties.setProperty("enable.auto.commit", " false");


        DataStream<Visit> source = env.addSource(new FlinkKafkaConsumer<>("flink_out", new SimpleStringSchema(), properties))
                .map(new MapFunction<String, Visit>() {
                    @Override
                    public Visit map(String value) throws Exception {
                        System.out.println(value);
                        JSONObject object = JSON.parseObject(value);
                        return new Visit(
                                object.getInteger("userId"),
                                object.getString("page"),
                                object.getString("action"),
                                object.getString("userActionTime")
                        );
                    }
                });

//        source.print();

        source.addSink(JdbcSink.sink(
                "insert into visit(userId, page, action, userActionTime) values (?,?,?,?)",
                (ps, value) -> {
                    ps.setInt(1, value.userId);
                    ps.setString(2, value.page);
                    ps.setString(3, value.action);
                    ps.setString(4,value.userActionTime);
                },
                JdbcExecutionOptions.builder()
                        .withBatchSize(1000)
                        .withBatchIntervalMs(200)
                        .withMaxRetries(5)
                        .build(),
                new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                        .withUrl("jdbc:mysql://192.168.62.34:3306/test")
                        .withDriverName("com.mysql.jdbc.Driver")
                        .withUsername("someUser")
                        .withPassword("somePassword")
                        .build()
                ));


        env.execute();

    }

    static class Visit {
        Integer userId;
        String page;
        String action;
        String userActionTime;

        public Visit() {
        }

        public Visit(Integer userId, String page, String action, String userActionTime) {
            this.userId = userId;
            this.page = page;
            this.action = action;
            this.userActionTime = userActionTime;
        }

        public Integer getUserId() {
            return userId;
        }

        public void setUserId(Integer userId) {
            this.userId = userId;
        }

        public String getPage() {
            return page;
        }

        public void setPage(String page) {
            this.page = page;
        }

        public String getAction() {
            return action;
        }

        public void setAction(String action) {
            this.action = action;
        }

        public String getUserActionTime() {
            return userActionTime;
        }

        public void setUserActionTime(String userActionTime) {
            this.userActionTime = userActionTime;
        }

        @Override
        public String toString() {
            return JSON.toJSONString(this);
        }
    }


}
